package com.zeng.ssm.service.support.async;

import com.zeng.ssm.api.support.idempotent.DistributedLockService;
import com.zeng.ssm.dal.mapper.AsyncTaskLogDao;
import com.zeng.ssm.dal.mapper.AsyncTaskRecordDao;
import com.zeng.ssm.dal.model.AsyncTaskLog;
import com.zeng.ssm.dal.model.AsyncTaskRecord;
import com.zeng.ssm.service.support.async.dto.AsyncTaskLogDto;
import com.zeng.ssm.service.support.async.dto.AsyncTaskRecordDto;
import com.zeng.ssm.util.DateUtil;
import com.zeng.ssm.util.TraceNoUtil;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;

import java.nio.charset.StandardCharsets;

/**
 * 异步任务Base类
 *
 * @author zengwanlin
 * @date 2023-04-03T19:25
 */
public abstract class BaseAsyncTaskExecutor {

    private Logger logger = LoggerFactory.getLogger(BaseAsyncTaskExecutor.class);

    @Autowired
    private AsyncTaskLogDao asyncTaskLogDao;

    @Autowired
    private AsyncTaskRecordDao asyncTaskRecordDao;

    @Autowired
    private DistributedLockService distributedLockService;

    /**
     * 是否已成功
     *
     * @param context
     * @return
     */
    protected boolean succeed(AsyncTaskContext context) {
        return false;
    }

    /**
     * 任务互斥
     * <p>
     * 为true时，单个类型的任务之间串行执行
     *
     * @return
     */
    protected boolean mutex() {
        return false;
    }

    /**
     * 执行任务
     *
     * @param context
     */
    protected final void execute(AsyncTaskContext context) {
        AsyncTaskRecordDto taskRecordDto = context.getTaskRecordDto();
        AsyncTaskLog logRecord = new AsyncTaskLog();
        AsyncTaskRecord record = new AsyncTaskRecord();
        record.setId(taskRecordDto.getId());
        record.setExecuteTimes(taskRecordDto.getExecuteTimes() + 1);

        final String lockKey = "ASYNC_LOCK_" + getLockKey(taskRecordDto);
        final String uniqueValue = RandomStringUtils.randomAlphanumeric(20);
        boolean lockSuc = distributedLockService.lock(lockKey, uniqueValue, taskRecordDto.getLockTtl() * 5);
        if (!lockSuc) {
            logger.info("任务ID[{}]锁定任务失败，跳过本次执行", taskRecordDto.getId());
            return;
        }

        try {
            AsyncTaskLog taskLog = new AsyncTaskLog();
            taskLog.setBusiDate(DateUtil.getDate8());
            taskLog.setRecordId(taskRecordDto.getId());
            taskLog.setHost(AsyncTaskExecutorLauncher.LOCAL_IP4_ADDRESS);
            taskLog.setStatus(AsyncTaskConstant.ASYNC_TASK_STATUS_0);
            taskLog.setTraceNo(TraceNoUtil.getTraceNo());
            taskLog.setCreateTime(DateUtil.getDateTime());
            asyncTaskLogDao.insertSelective(taskLog);
            logRecord.setId(taskLog.getId());

            AsyncTaskLogDto taskLogDto = new AsyncTaskLogDto();
            BeanUtils.copyProperties(taskLog, taskLog);
            context.setTaskLog(taskLogDto);

            // 执行任务
            doExecute(context);

            // 更新任务记录
            record.setStatus(AsyncTaskConstant.ASYNC_TASK_STATUS_1);
            record.setLockOwner(AsyncTaskExecutorLauncher.DEFAULT_LOCK_OWNER);
            record.setUpdateTime(DateUtil.getDateTime());
            asyncTaskRecordDao.updateByPrimaryKeySelective(record);
            // 更新日志记录
            logRecord.setStatus(AsyncTaskConstant.ASYNC_TASK_STATUS_1);
            logRecord.setRetMsg(safeRetMsg(context.getRetMsg()));
            logRecord.setUpdateTime(DateUtil.getDateTime());
            asyncTaskLogDao.updateByPrimaryKeySelective(logRecord);

            context.setStatus(record.getStatus());
        } catch (Throwable th) {
            // 更新任务记录
            logger.error(th.getMessage(), th);
            if (taskRecordDto.getExecuteTimes() < context.getTaskConfigDto().getMaxExecuteTimes()) {
                record.setStatus(AsyncTaskConstant.ASYNC_TASK_STATUS_2);
            } else {
                record.setStatus(AsyncTaskConstant.ASYNC_TASK_STATUS_3);
            }
            record.setUpdateTime(DateUtil.getDateTime());
            asyncTaskRecordDao.updateByPrimaryKeySelective(record);
            // 更新日志记录
            logRecord.setStatus(record.getStatus());
            logRecord.setRetMsg(safeRetMsg(ExceptionUtils.getMessage(th)));
            logRecord.setUpdateTime(DateUtil.getDateTime());
            asyncTaskLogDao.updateByPrimaryKeySelective(logRecord);
            context.setStatus(record.getStatus());
        } finally {
            distributedLockService.unlock(lockKey, uniqueValue);
        }
    }

    /**
     * 已触发成功
     *
     * @param context
     */
    protected final void alreadySucceed(AsyncTaskContext context) {
        AsyncTaskRecordDto taskRecordDto = context.getTaskRecordDto();

        AsyncTaskRecord record = new AsyncTaskRecord();
        record.setId(taskRecordDto.getId());
        record.setExecuteTimes(taskRecordDto.getExecuteTimes() + 1);
        record.setLockOwner(AsyncTaskExecutorLauncher.DEFAULT_LOCK_OWNER);
        record.setUpdateTime(DateUtil.getDateTime());
        record.setStatus(AsyncTaskConstant.ASYNC_TASK_STATUS_1);
        asyncTaskRecordDao.updateByPrimaryKeySelective(record);

        AsyncTaskLog taskLog = new AsyncTaskLog();
        taskLog.setBusiDate(DateUtil.getDate8());
        taskLog.setRecordId(taskRecordDto.getId());
        taskLog.setHost(AsyncTaskExecutorLauncher.LOCAL_IP4_ADDRESS);
        taskLog.setTraceNo(TraceNoUtil.getTraceNo());
        taskLog.setStatus(AsyncTaskConstant.ASYNC_TASK_STATUS_1);
        taskLog.setRetMsg("已操作成功");
        taskLog.setCreateTime(DateUtil.getDate8());
        taskLog.setUpdateTime(DateUtil.getDate8());
        asyncTaskLogDao.insertSelective(taskLog);
    }

    /**
     * 执行异步任务
     *
     * @param context
     */
    protected abstract void doExecute(AsyncTaskContext context);

    /**
     * 安全长度的错误信息
     *
     * @param retMsg
     * @return
     */
    private String safeRetMsg(String retMsg) {
        if (StringUtils.isBlank(retMsg)) {
            return null;
        }
        // 数据库byte长度
        int allowByteLen = 512;
        byte[] msgBytes = retMsg.getBytes(StandardCharsets.UTF_8);
        if (msgBytes.length <= allowByteLen) {
            return retMsg;
        }
        return new String(msgBytes, 0, allowByteLen, StandardCharsets.UTF_8);
    }

    /**
     * 获取锁Key
     *
     * @param recordDto
     * @return
     */
    private String getLockKey(AsyncTaskRecordDto recordDto) {
        if (mutex()) {
            // 任务互斥
            return "MUTEX_" + recordDto.getTaskCode();
        }
        if (StringUtils.isNotEmpty(recordDto.getDetailId())) {
            return "DET_ID_" + recordDto.getDetailId();
        }
        if (StringUtils.isNotEmpty(recordDto.getContractId())) {
            return "CON_ID_" + recordDto.getContractId();
        }
        if (StringUtils.isNotEmpty(recordDto.getResourceId())) {
            if (StringUtils.isNotBlank(recordDto.getResourceSubId())) {
                return "RES_" + recordDto.getTaskCode() + AsyncTaskConstant.HORIZONTAL_LINE + recordDto.getResourceId() + AsyncTaskConstant.HORIZONTAL_LINE + recordDto.getResourceSubId();
            } else {
                return "RES_" + recordDto.getTaskCode() + AsyncTaskConstant.HORIZONTAL_LINE + recordDto.getResourceId();
            }
        }
        return "ID_" + recordDto.getId().toString();
    }

}
